package main

import (
	"context"
	"fmt"
	"sync"
	"time"
)

//basic mode
//Do() when Finished() Next()
type KeeperNode interface {
	SetNext(KeeperNode, KeeperNode) error
	SetPrevious(KeeperNode) error
	Next() (KeeperNode, context.Context) //call when finished
	Meta() map[string]string             // id ,type,version,etc...
	Finished() bool
	Result() (interface{}, error) //as always
	Do(context.Context)           //block
}

type StateGroup struct {
	mu       sync.RWMutex
	count    int
	runnig   map[int]int64
	finished map[int]int64
	ntf      chan int
	errs     []error
}

func NewStateGroup(count int) *StateGroup {
	return &StateGroup{
		runnig:   make(map[int]int64),
		finished: make(map[int]int64),
		ntf:      make(chan int, count),
		count:    count,
	}
}

func (sg *StateGroup) Signal() <-chan int {
	return sg.ntf
}

func (sg *StateGroup) OnError(e error) {
	sg.mu.Lock()
	defer sg.mu.Unlock()
	sg.errs = append(sg.errs, e)
}

func (sg *StateGroup) OnFinish(i int) {
	sg.mu.Lock()
	defer sg.mu.Unlock()
	sg.finished[i] = time.Now().Unix()
	delete(sg.runnig, i)
	sg.ntf <- i
}

func (sg *StateGroup) Errors() int {
	sg.mu.RLock()
	defer sg.mu.RUnlock()
	return len(sg.errs)
}
func (sg *StateGroup) Finished() int {
	sg.mu.RLock()
	defer sg.mu.RUnlock()
	var r []int
	for i := range sg.finished {
		r = append(r, i)
	}
	return len(r)
}

func (sg *StateGroup) Running() int {
	sg.mu.RLock()
	defer sg.mu.RUnlock()
	var r []int
	for i := range sg.runnig {
		r = append(r, i)
	}
	return len(r)
}

func (sg *StateGroup) OnStart(i int) {
	sg.mu.Lock()
	defer sg.mu.Unlock()
	delete(sg.finished, i)
	sg.runnig[i] = time.Now().Unix()
}

type SerialNode struct {
	nodes   []KeeperNode
	errNext KeeperNode
	next    KeeperNode
	pre     KeeperNode
	id      string
	ctx     context.Context

	state StateGroup
	ret   interface{}
}

type ParallelNode struct {
	nodes []KeeperNode
	state StateGroup

	errNext KeeperNode
	next    KeeperNode
	pre     KeeperNode

	id  string
	ctx context.Context

	errorTolerance int
	parallelSize   int
}

func (p *SerialNode) SetNext(n, e KeeperNode) error {
	p.next = n
	p.errNext = e
	return nil
}

func (p *SerialNode) SetPrevious(n KeeperNode) error {
	p.pre = n
	return nil
}

func (p *SerialNode) Next() (KeeperNode, context.Context) {
	if p.state.Errors() == 0 {
		return p.next, p.ctx
	}
	return p.errNext, p.ctx
}
func (p *SerialNode) Meta() map[string]string {
	return map[string]string{"id": p.id}
}
func (p *SerialNode) Finished() bool {
	return p.state.Finished() == len(p.nodes) || p.state.Errors() != 0
}
func (p *SerialNode) Result() (interface{}, error) {
	if p.state.Errors() == 0 {
		return nil, nil
	}
	return nil, fmt.Errorf("errors in serial node")
}
func (p *SerialNode) Do(ctx context.Context) {
	p.ctx = ctx
	for i, n := range p.nodes {
		p.state.OnStart(i)
		n.Do(context.WithValue(ctx, "parent.node.meta", p.Meta()))
		p.state.OnFinish(i)
		if _, err := n.Result(); err != nil {
			p.state.OnError(err)
			break
		}
	}
}

func (p *ParallelNode) SetNext(n, e KeeperNode) error {
	p.next = n
	p.errNext = e
	return nil
}

func (p *ParallelNode) SetPrevious(n KeeperNode) error {
	p.pre = n
	return nil
}
func (p *ParallelNode) Next() (n KeeperNode, ctx context.Context) { //call when finished
	if !p.Finished() {
		return nil, nil
	}
	if _, err := p.Result(); err != nil {
		return p.errNext, p.ctx
	}
	return p.next, nil
}

func (p *ParallelNode) Meta() map[string]string {
	return map[string]string{"id": p.id}
}

func (p *ParallelNode) Finished() bool {
	return p.state.Finished() == len(p.nodes)
}

func (p *ParallelNode) Result() (interface{}, error) {
	if l := p.state.Errors(); l > p.errorTolerance {
		return nil, fmt.Errorf("errors %d exceed %d", l, p.errorTolerance)
	}
	return nil, nil
}

func ExecDo(ctx context.Context, p *ParallelNode, n KeeperNode, i int) {
	p.state.OnStart(i)
	n.Do(context.WithValue(ctx, "parent.node.meta", p.Meta()))
	_, err := n.Result()
	if err != nil {
		p.state.OnError(err)
	}
	p.state.OnFinish(i)
}

func (p *ParallelNode) Do(ctx context.Context) {
	p.ctx = ctx
	for i, n := range p.nodes {
		if count := p.state.Running(); count > p.parallelSize {
			<-p.state.Signal()
		}
		/*go func() {
			p.state.OnStart(i)
			n.Do(context.WithValue(ctx, "parent.node.meta", p.Meta()))
			_, err := n.Result()
			if err != nil {
				p.state.OnError(err)
			}
			p.state.OnFinish(i)
		}()*/
		go ExecDo(ctx, p, n, i)
	}
	for p.state.Running() != 0 {
		<-p.state.Signal()
	}
}
